package org.sn.jdish.spark.sql;

import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.hive.HiveContext;

/**
 * 读取Hive数据源
 * 
 * @author snzigod@hotmail.com
 */
public class HiveDataSource {

	public static void main(String[] args) {
		/**
		 * SparkConf:第一步创建一个SparkConf，在这个对象里面可以设置允许模式Local Standalone yarn
		 * AppName(可以在Web UI中看到) 还可以设置Spark运行时的资源要求
		 */
		SparkConf sparkConf = new SparkConf().setAppName("sparkSql").setMaster("local");

		/**
		 * 基于SparkConf的对象可以创建出来一个SparkContext Spark上下文
		 * SparkContext是通往集群的唯一通道，SparkContext在创建的时候还会创建任务调度器
		 */
		JavaSparkContext javaSparkContext = new JavaSparkContext(sparkConf);

		readJson(javaSparkContext);

		javaSparkContext.close();
	}

	public static void readJson(JavaSparkContext javaSparkContext) {

		/*
		 * 0.把hive里面的hive-site.xml放到spark/conf目录下 1.启动Mysql 2.启动HDFS 3.启动Hive ./hive
		 * 4.初始化HiveContext 5.打包运行
		 * 
		 * ./bin/spark-submit --master yarn-cluster --class
		 * com.huangyueran.spark.sql.HiveDataSource /root/spark_hive_datasource.jar
		 * ./bin/spark-submit --master yarn-client --class
		 * com.huangyueran.spark.sql.HiveDataSource /root/spark_hive_datasource.jar
		 */

		// 创建HiveContext，注意，这里，它接收的是SparkContext作为参数，不是JavaSparkContext，其实也可以使用JavaSparkContext，只不过内部也是做了sc.sc()的操作
		HiveContext hiveContext = new HiveContext(javaSparkContext.sc());
		DataFrame person = hiveContext.sql("select * from psn1");
		person.show();

		List<Row> list = person.javaRDD().collect();
		System.out.println("=============================================================");
		for (Row r : list) {
			System.out.println(r);
		}
		System.out.println("=============================================================");
	}

}
